Add David AI redelivered MFA pipeline from draco#2186
Conversation
Sync current pipeline (RAM-by-session stages, glued-OOV heuristic detector, lexicon build, cluster submission scripts, ffmpeg timeout hardening) from the draco working copy.
timofeeva16tanya
requested review from
sarahyurick
and removed request for
a team
Greptile SummaryThis PR adds the David AI redelivered MFA pipeline from the draco working copy as a new tutorial under
Confidence Score: 4/5Safe to merge with one fix: the WAV variant's apply_white_noise_in_pause_intervals should add the same empty-pause short-circuit present in the opus variant to avoid unnecessary ffmpeg round-trips that can fail on dense-speech sessions. The WAV pipeline's apply_white_noise_in_pause_intervals always decodes and re-encodes audio even when pause_intervals is empty, while the opus version correctly short-circuits to a file copy. On sessions where speech covers the full recording, this causes an avoidable ffmpeg call that can propagate TimeoutExpired and mark the session as failed when it should have succeeded. tutorials/audio/david_ai_redelivered_mfa/wav/david_ai_common.py — the apply_white_noise_in_pause_intervals function needs the same empty-pause guard that the opus counterpart in opus/david_ai_common.py already has. Important Files Changed
Reviews (3): Last reviewed commit: "new_vwersion" | Re-trigger Greptile |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| start = stage_names.index(stage_name) | ||
| except ValueError: | ||
| return | ||
| for name in stage_names[stage:]: |
There was a problem hiding this comment.
NameError on every call: the local variable is start but the slice uses the undefined name stage. Any caller of clear_stage_done_from will crash with NameError: name 'stage' is not defined at runtime.
| for name in stage_names[stage:]: | |
| for name in stage_names[start:]: |
| for session_dir in sessions | ||
| ] |
There was a problem hiding this comment.
sessions (all sessions) instead of todo_sessions (only the ones that need processing). The todo_sessions filter was correctly computed and logged, but is never used here. Every already-finished session is submitted to the process pool, where process_session_ram parses manifests and re-checks outputs before returning skipped=True — a significant waste at scale (e.g. a 200-node run re-processes every completed session on restart).
| for session_dir in sessions | |
| ] | |
| for session_dir in todo_sessions | |
| ] |
| def recording_textgrid_paths(textgrid_dir: Path, recording_id: str) -> list[Path]: | ||
| ordinary = recording_textgrid_path(textgrid_dir, recording_id, variant="ordinary") | ||
| if ordinary.is_file(): | ||
| return [ordinary] | ||
| fb_path = recording_textgrid_path(textgrid_dir, recording_id, variant="fb") | ||
| if fb_path.is_file(): | ||
| return [ordinary, fb_path] | ||
| return [ordinary] |
There was a problem hiding this comment.
ordinary does not exist as a file but fb_path does, the function returns [ordinary, fb_path] — a list that includes the non-existent ordinary path. build_recording_rttm_lines iterates that list and calls parse_textgrid_words(ordinary), which raises an exception that is silently swallowed; only the fb path is parsed. The intent appears to be returning just the fb path (or both when both exist), but the current code returns a stale, missing path.
| def recording_textgrid_paths(textgrid_dir: Path, recording_id: str) -> list[Path]: | |
| ordinary = recording_textgrid_path(textgrid_dir, recording_id, variant="ordinary") | |
| if ordinary.is_file(): | |
| return [ordinary] | |
| fb_path = recording_textgrid_path(textgrid_dir, recording_id, variant="fb") | |
| if fb_path.is_file(): | |
| return [ordinary, fb_path] | |
| return [ordinary] | |
| def recording_textgrid_paths(textgrid_dir: Path, recording_id: str) -> list[Path]: | |
| ordinary = recording_textgrid_path(textgrid_dir, recording_id, variant="ordinary") | |
| fb_path = recording_textgrid_path(textgrid_dir, recording_id, variant="fb") | |
| if ordinary.is_file() and fb_path.is_file(): | |
| return [ordinary, fb_path] | |
| if ordinary.is_file(): | |
| return [ordinary] | |
| if fb_path.is_file(): | |
| return [fb_path] | |
| return [ordinary] |
| from pathlib import Path | ||
| from typing import TypeVar | ||
|
|
||
| MFA_ROOT_DIR_DEFAULT = "/home/ttimofeeva/MFA_models" |
There was a problem hiding this comment.
MFA_ROOT_DIR will get a FileNotFoundError when resolving models, with a confusing path pointing to ttimofeeva's home. A generic fallback like ~/MFA_models is more appropriate.
| MFA_ROOT_DIR_DEFAULT = "/home/ttimofeeva/MFA_models" | |
| MFA_ROOT_DIR_DEFAULT = "~/MFA_models" |
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
| try: | ||
| result = subprocess.run(align_cmd, capture_output=True, text=True, env=mfa_env) | ||
| except OSError as exc: | ||
| logger.error("%s: mfa align failed to start: %s", recording_id, exc) | ||
| mfa_failed_globally = True | ||
| detail = str(exc) |
There was a problem hiding this comment.
mfa align subprocess
subprocess.run(align_cmd, …) has no timeout= argument. The codebase already documents this exact failure mode: FFMPEG_TIMEOUT_S = 600 exists precisely because "a wedged ffmpeg (e.g. an internal futex deadlock seen when many run in a worker pool) blocks its caller forever and hangs the whole shard." MFA runs as another pooled subprocess with the same risk — if mfa align hangs (e.g. SQLite or pynini lock contention), the entire process pool worker blocks indefinitely with no recovery path. The outer try/except Exception in align_recording would catch a subprocess.TimeoutExpired and return RecordingAlignResult(ok=False), so adding a timeout here is safe and consistent.
| try: | ||
| start = stage_names.index(stage_name) | ||
| except ValueError: | ||
| return | ||
| for name in stage_names[stage:]: |
There was a problem hiding this comment.
NameError in clear_stage_done_from: stage vs start
The loop slices on stage, but the variable assigned by list.index() is named start. Any caller will crash with NameError: name 'stage' is not defined at runtime.
| try: | |
| start = stage_names.index(stage_name) | |
| except ValueError: | |
| return | |
| for name in stage_names[stage:]: | |
| try: | |
| start = stage_names.index(stage_name) | |
| except ValueError: | |
| return | |
| for name in stage_names[start:]: |
1 participant
Sync current pipeline (RAM-by-session stages, glued-OOV heuristic detector, lexicon build, cluster submission scripts, ffmpeg timeout hardening) from the draco working copy.
Description
Usage
# Add snippet demonstrating usageChecklist